package top.lunarye.support.delayMessage;

import com.alibaba.fastjson.JSON;
import io.netty.util.HashedWheelTimer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import top.lunarye.common.constant.RocketDelayConstants;
import top.lunarye.support.utils.DelayLevelCalculateUtils;
import top.lunarye.support.utils.RedisUtils;
import top.lunarye.support.utils.TimeWheelFactory;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.Instant;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author lunar
 * @since 2022-07-12
 */
@Slf4j
@Component
public class DelayMessageProducer extends DefaultMQProducer {

    private DefaultMQPushConsumer defaultMQPushConsumer;

    @Resource
    private SendRealMessage sendRealMqTask;

    @Resource
    private RedisUtils redisUtils;

    public DelayMessageProducer() {
        super();
    }

    public void initProducer(String producerGroup, String retryTime, String mqUrl) {
        this.setProducerGroup(producerGroup);
        this.setRetryTimesWhenSendFailed(Integer.parseInt(retryTime));
        this.setNamesrvAddr(mqUrl);
    }

    /**
     * 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h rocketMQ自动支持18个级别 等级全部转化为秒
     *
     * 延时小于1分钟的 不支持，可以设置强制参数来支持，但是会有误差
     */
    public SendResult sendDelay(Message msg, Date startSendTime) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        if (startSendTime == null) {
            return super.send(msg);
        }
        long l = Duration.between(Instant.now(), startSendTime.toInstant()).getSeconds();
        //如果不等于0,说明设置了延时等级，直接用rocketMQ支持的发送
        // 在需要延迟的时间小于一分钟的时候，就使用netty中的时间轮算法来实现本机的延迟
        if (l <= RocketDelayConstants.TIME_OUT) {
            storeInRedis(msg, startSendTime);
            HashedWheelTimer instance = TimeWheelFactory.getInstance();
            CountDownLatch countDownLatch = new CountDownLatch(1);
//            SendRealMessage sendRealMqTask = new SendRealMessage();
            sendRealMqTask.setDelayMessageProducer(this);
            sendRealMqTask.setMessage(msg);
            sendRealMqTask.setCountDownLatch(countDownLatch);
            instance.newTimeout(sendRealMqTask, l < 0 ? 1 : l, TimeUnit.SECONDS);
            countDownLatch.await();
            return sendRealMqTask.getResult();
        } else { // 需要延迟的时间超过一分钟的情况下，使用rocketmq中的自带的延迟队列来拼凑出自定义的延迟时间
            Integer level = DelayLevelCalculateUtils.calculateDefault(l);
            fillMessage(msg, level, startSendTime);
            return super.send(msg);
        }
    }

    /**
     * 在消息进入时间轮中倒计时的同时，将该消息存储进入redis中防止消息丢失
     * @param msg 消息的主题
     * @param startSendTime 消息将要发送的时间
     */
    private void storeInRedis(Message msg, Date startSendTime) {
        Map<String, Object> msgMap = new HashMap<>();
        msgMap.put("topic", msg.getTopic());
        msgMap.put("tags", msg.getTags());
        msgMap.put("body", new String(msg.getBody()));
        redisUtils.zSet(RocketDelayConstants.REDIS_STORE_ZSET, JSON.toJSONString(msgMap), startSendTime.getTime());
    }

    /**
     * 在启动producer的时候，同时启动一个消费者来进行自定义时间延迟消息的消费
     * @throws MQClientException
     */
    @Override
    public void start() throws MQClientException {
        super.start();
        initConsumer();
    }

    /**
     * 初始化消费者
     * @throws MQClientException
     */
    private void initConsumer() throws MQClientException {
        defaultMQPushConsumer = new DefaultMQPushConsumer(super.getProducerGroup());
        defaultMQPushConsumer.setNamesrvAddr(super.getNamesrvAddr());
        defaultMQPushConsumer.setMessageListener(new MessageListener(this));
        defaultMQPushConsumer.subscribe(RocketDelayConstants.PROXY_TOPIC, "*");
        defaultMQPushConsumer.start();
    }

    /**
     * 在终结生产者的同时将消费者也终结
     */
    @Override
    public void shutdown() {
        super.shutdown();
        if (defaultMQPushConsumer != null) {
            defaultMQPushConsumer.shutdown();
        }
    }

    /**
     * 填充消息
     *
     * @param msg 消息体
     * @param level 延时等级
     * @param startSendTime 发送时间
     */
    private void fillMessage(Message msg, Integer level, Date startSendTime) {
        msg.putUserProperty(RocketDelayConstants.TIMES, String.valueOf(startSendTime.getTime() / 1000 - DelayLevelCalculateUtils.get(level - 1)));

        String topic = msg.getProperty(RocketDelayConstants.ORIGINAL_TOPIC);
        if (StringUtils.isBlank(topic)) {
            msg.putUserProperty(RocketDelayConstants.ORIGINAL_TOPIC, msg.getTopic());
        }
        String tag = msg.getProperty(RocketDelayConstants.ORIGINAL_TAG);
        if (StringUtils.isBlank(tag)) {
            if (StringUtils.isNotBlank(msg.getTags())) {
                msg.putUserProperty(RocketDelayConstants.ORIGINAL_TAG, msg.getTags());
            }
        }
        String keys = msg.getProperty(RocketDelayConstants.ORIGINAL_KEY);
        if (StringUtils.isBlank(keys)) {
            if (StringUtils.isNotBlank(msg.getKeys())) {
                msg.putUserProperty(RocketDelayConstants.ORIGINAL_KEY, msg.getKeys());
            }
        }

        String property = msg.getProperty(RocketDelayConstants.ORIGINAL_UUID);
        String uuid = UUID.randomUUID().toString().replaceAll("-", "");
        if (StringUtils.isBlank(property)) {
            msg.putUserProperty(RocketDelayConstants.ORIGINAL_UUID, uuid);
        }
        msg.setDelayTimeLevel(level);
        msg.setTopic(RocketDelayConstants.PROXY_TOPIC);
        log.info("消息uuid {} 开发发送时间为{},延时等级本次建议为{}", uuid, String.format("%tF %<tT", new Date()), level);
    }
}
